package com.rankeiot.core.sqltoy;


import org.sagacity.sqltoy.callback.StreamResultHandler;
import org.sagacity.sqltoy.dao.SqlToyLazyDao;
import org.sagacity.sqltoy.model.QueryExecutor;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;

public class StreamDataSource<T> {

    private Consumer<Consumer<T>> callback;
    private Runnable finishHandler = () -> {
    };

    public StreamDataSource(SqlToyLazyDao dao, QueryExecutor qe) {
        callback = (ch) -> {
            dao.fetchStream(qe, new ConsumerStreamResultHandler(ch));
            finishHandler.run();
        };
    }

    public StreamDataSource(Collection<T> collection) {
        callback = (ch) -> {
            for (T obj : collection) {
                ch.accept(obj);
            }
            finishHandler.run();
        };
    }

    public StreamDataSource(T[] collection) {
        callback = (ch) -> {
            for (T obj : collection) {
                ch.accept(obj);
            }
            finishHandler.run();
        };
    }
    public StreamDataSource(int count){
        callback = (ch) -> {
            for (Integer i=0;i<count;i++) {
                ch.accept((T)i);
            }
            finishHandler.run();
        };
    }
    public StreamDataSource(Consumer<Consumer<T>> callback) {
        this.callback = (ch) -> {
            callback.accept(ch);
            finishHandler.run();
        };
    }

    public void foreach(Consumer<T> consumer) {
        callback.accept(consumer);
    }

    public void onFinish(Runnable runnable) {
        this.finishHandler = runnable;
    }

    public static StreamDataSource<Map<String, Object>> merge(Map<String, StreamDataSource> sdcs) {

        if(sdcs.size()==1){
            String name=sdcs.keySet().iterator().next();
            StreamDataSource target=sdcs.values().iterator().next();
            return new StreamDataSource<Map<String,Object>>(callback -> {
               target.foreach(c->{
                   Map<String,Object> nv=new HashMap<>();
                   nv.put(name,c);
                   callback.accept(nv);
               });
            });
        }
        //ExecutorService executor = Executors.newFixedThreadPool(1);
        StreamDataSource newConn = new StreamDataSource((Consumer<Consumer>) callback -> {
            ExecutorService executor = Executors.newFixedThreadPool(sdcs.size());
            final Counter ct = new Counter(sdcs.size());
            ct.onData(dt -> callback.accept(dt));
            Future[] features=new Future[sdcs.size()];
            int idx=0;
            for (Map.Entry<String, StreamDataSource> entry : sdcs.entrySet()) {
                String name = entry.getKey();
                StreamDataSource current = entry.getValue();
                current.onFinish(ct::subTotal);
                Future feature = executor.submit(() -> {
                    current.foreach((e) -> {
                        ct.put(name, e);
                    });
                });
                features[idx]=feature;
                idx++;
            }
            executor.shutdown();
            try {
               executor.awaitTermination(30,TimeUnit.MINUTES);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }


           // executor.awaitTermination(10, TimeUnit.MINUTES);

        });
        return newConn;
    }

    private static class Counter {
        private int total = 0;
        private int current = 0;
        private Map<String, Object> data;
        private Consumer<Map> consumer;

        public Counter(int total) {
            this.total = total;
            data = new HashMap<>();
        }

        public void subTotal() {
            synchronized (this) {
                total--;
                if (current == total&&total>0) {
                    next();
                    this.notifyAll();
                }
            }
        }

        public void put(String key, Object value) {

            synchronized (this) {
                data.put(key, value);
                current++;
                if(total==1){
                    next();
                }else if (current == total) {
                    next();
                    this.notifyAll();
                } else {
                    this.waitNext();
                }
            }
        }
        private void next(){
            consumer.accept(data);
            data = new HashMap<>();
            current = 0;
        }
        private void waitNext() {
            try {
                this.wait();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        public void onData(Consumer<Map> consumer) {
            this.consumer = consumer;
        }
    }

    public static class ConsumerStreamResultHandler<T> implements StreamResultHandler {
        private Consumer<T> consumer;

        public ConsumerStreamResultHandler(Consumer<T> consumer) {
            this.consumer = consumer;
        }

        @Override
        public void consume(Object row, int rowIndex) {
            consumer.accept((T) row);
        }
    }
}
